package com.chenjj.bigdata.flink.sql;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 通过SQL API的方式统计socket输入的单词个数
 *
 * 用法e.g
 *
 * 1.在192.168.152.134上启动一个nc服务
 * nc -l 9000
 *
 * 2.启动本示例，或者将本示例导成jar包，在finlk服务上运行，host=192.168.152.134 port=9000
 *
 * 3.在192.168.152.134的nc上输出单词
 *
 * 4.本示例会统计期间你输入的单词个数
 */
public class WordCountBySocket {

    static Logger logger  = LoggerFactory.getLogger(WordCountBySocket.class);

    public static void main(String[] args) throws Exception {
        // the port to connect to
        final int port;
        final  String host;
        try {
            final ParameterTool params = ParameterTool.fromArgs(args);
            port = params.getInt("port");
            host = params.get("host");
        } catch (Exception e) {
            System.err.println("No port specified . Please run 'WordCountBySocket  --host <host> --port <port>'");
            return;
        }
        logger.info("host={},port={}",host,port);

        // get the execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // get input data by connecting to the socket
        DataStream<String> input = env.socketTextStream(host,port, "\n");

        //清洗数据
        SingleOutputStreamOperator<WordWithCount> ds =  input.flatMap(new FlatMapFunction<String, WordWithCount>() {
            @Override
            public void flatMap(String value, Collector<WordWithCount> out) {
                for (String word : value.split("\\s")) {
                    out.collect(new WordWithCount(word, 1L));
                }
            }
        });

        //register table
        tableEnv.registerDataStream("wordcountTable",ds,"word,count");

        tableEnv.executeSql("SELECT word,sum(`count`) as c from wordcountTable group by word").print();

        env.setParallelism(2);
        env.execute("WordCountBySocket");

    }

    // Data type for words with count
    public static class WordWithCount {

        public String word;
        public long count;

        public WordWithCount() {}

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }

}
